package io.github.wanggit.antrpc.client.spring;

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.wanggit.antrpc.IRpcClients;
import io.github.wanggit.antrpc.RpcClient;
import io.github.wanggit.antrpc.client.monitor.IRpcCallLogHolder;
import io.github.wanggit.antrpc.client.spring.exception.ResultWasNullException;
import io.github.wanggit.antrpc.client.spring.exception.ServiceProviderOccurredException;
import io.github.wanggit.antrpc.client.spring.utils.StackTraceUtil;
import io.github.wanggit.antrpc.client.zk.exception.InterfaceProviderNotFoundException;
import io.github.wanggit.antrpc.client.zk.zknode.INodeHostContainer;
import io.github.wanggit.antrpc.client.zk.zknode.NodeHostEntity;
import io.github.wanggit.antrpc.commons.bean.*;
import io.github.wanggit.antrpc.commons.bean.error.RpcResponseError;
import io.github.wanggit.antrpc.commons.breaker.IResilience4jCircuitBreakerContainer;
import io.github.wanggit.antrpc.commons.codec.serialize.ISerializerHolder;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.future.ReadClientFuture;
import io.github.wanggit.antrpc.commons.rate.IResilience4jRateLimiter;
import io.github.wanggit.antrpc.commons.utils.AntRpcClassUtils;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;

import java.lang.reflect.Method;
import java.util.List;

@Slf4j
public class CglibMethodInterceptor implements MethodInterceptor {

    private final IRpcCallLogHolder rpcCallLogHolder;
    private final IOnFailHolder onFailHolder;
    private final IResilience4jCircuitBreakerContainer circuitBreakerContainer;
    private final IResilience4jRateLimiter clientResilience4jRateLimiterContainer;
    private final IRpcClients rpcClients;
    private final ISerializerHolder serializerHolder;
    private final INodeHostContainer nodeHostContainer;

    CglibMethodInterceptor(
            IRpcCallLogHolder rpcCallLogHolder,
            IOnFailHolder onFailHolder,
            IResilience4jCircuitBreakerContainer circuitBreakerContainer,
            IResilience4jRateLimiter clientResilience4jRateLimiterContainer,
            IRpcClients rpcClients,
            ISerializerHolder serializerHolder,
            INodeHostContainer nodeHostContainer) {
        this.rpcCallLogHolder = rpcCallLogHolder;
        this.onFailHolder = onFailHolder;
        this.circuitBreakerContainer = circuitBreakerContainer;
        this.clientResilience4jRateLimiterContainer = clientResilience4jRateLimiterContainer;
        this.rpcClients = rpcClients;
        this.serializerHolder = serializerHolder;
        this.nodeHostContainer = nodeHostContainer;
    }

    @Override
    public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy)
            throws Throwable {
        if (method.getDeclaringClass().equals(Object.class)) {
            return proxy.invokeSuper(obj, args);
        }
        Class<?> anInterface = internalDoManyLoggerAndGetInterface(obj, method);
        String interfaceClassName = anInterface.getName();
        String previousStackTraceElement =
                StackTraceUtil.getRealPreviousStackTraceElement(
                        Thread.currentThread().getStackTrace());
        RegisterBeanMethod registerBeanMethod = RegisterBeanHelper.getRegisterBeanMethod(method);
        // RegisterBeanMethod中的class是实现类，替换为接口类
        registerBeanMethod.setClassName(interfaceClassName);
        RpcCallLog rpcCallLog =
                initRpcCallLog(
                        registerBeanMethod, interfaceClassName, args, previousStackTraceElement);
        NodeHostEntity hostEntity = null;
        try {
            hostEntity = chooseNodeHostEntity(interfaceClassName, registerBeanMethod.toString());
        } catch (InterfaceProviderNotFoundException e) {
            if (log.isErrorEnabled()) {
                log.error(e.getMessage(), e);
            }
            doRpcCallLogWhenError(hostEntity, rpcCallLog, e);
        }
        if (null == hostEntity) {
            return null;
        }
        NodeHostEntity targetHostEntity = hostEntity;
        RpcRequestBean rpcRequestBean =
                createRpcRequestBean(interfaceClassName, registerBeanMethod, args);
        String callUniqueKey = getCallLogKey(rpcRequestBean, hostEntity);
        // rate limiting
        return clientResilience4jRateLimiterContainer.run(
                registerBeanMethod,
                () -> {
                    // circuit breaker
                    return circuitBreakerContainer.run(
                            interfaceClassName,
                            callUniqueKey,
                            () -> {
                                RpcProtocol rpcProtocol = createRpcProtocol(rpcRequestBean);
                                return doInternalSendWhenCircuitBreakerClosed(
                                        targetHostEntity, rpcRequestBean, rpcCallLog, rpcProtocol);
                            },
                            throwable ->
                                    doRpcCallLogWhenError(targetHostEntity, rpcCallLog, throwable),
                            throwable -> {
                                CircuitBreaker.State state =
                                        circuitBreakerContainer.getState(callUniqueKey);
                                if (CircuitBreaker.State.OPEN == state) {
                                    return doOnFailHolderWhenError(
                                            method, args, anInterface, throwable);
                                }
                                return null;
                            });
                },
                throwable -> doRpcCallLogWhenError(targetHostEntity, rpcCallLog, throwable),
                throwable -> doOnFailHolderWhenError(method, args, anInterface, throwable));
    }

    public Class<?> internalDoManyLoggerAndGetInterface(Object obj, Method method) {
        List<Class<?>> allInterfaces = AntRpcClassUtils.getAllInterfaces(obj);
        if (allInterfaces.isEmpty()) {
            String message =
                    "The "
                            + obj.getClass().getName()
                            + " object must implement at least one interface.";
            if (log.isErrorEnabled()) {
                log.error(message);
            }
            throw new IllegalStateException(message);
        }
        if (log.isDebugEnabled()) {
            log.debug("class=" + obj.getClass() + ", allInterfaces=" + allInterfaces.toString());
        }
        Class<?> anInterface = allInterfaces.get(0);
        if (log.isDebugEnabled()) {
            log.debug("class=" + obj.getClass() + ", choosed interface=" + anInterface.getName());
        }
        if (log.isDebugEnabled()) {
            log.debug(anInterface.getName() + "#" + method.getName());
        }
        return anInterface;
    }

    /**
     * 当发生限流与熔断时，处理消费端的OnFailHolder
     *
     * @param method method
     * @param args args
     * @param anInterface interfaceClass
     * @param throwable occured exception
     * @return default on fail
     */
    private Object doOnFailHolderWhenError(
            Method method, Object[] args, Class<?> anInterface, Throwable throwable) {
        try {
            return onFailHolder.doOnFail(anInterface, method, args);
        } catch (OnFailHolder.NotFoundOnFailHandlerException e) {
            if (log.isErrorEnabled()) {
                log.error(throwable.getMessage() + "[" + e.getMessage() + "]", throwable);
            }
        }
        return null;
    }

    /**
     * 当异常发生时记录日志
     *
     * @param hostEntity hostEntity
     * @param rpcCallLog rpcCallLog entity
     * @param throwable occured exception
     */
    private void doRpcCallLogWhenError(
            NodeHostEntity hostEntity, RpcCallLog rpcCallLog, Throwable throwable) {
        if (log.isErrorEnabled()) {
            log.error(throwable.getMessage(), throwable);
        }
        if (null != rpcCallLog) {
            if (null != hostEntity) {
                rpcCallLog.setTargetIp(hostEntity.getIp());
                rpcCallLog.setTargetPort(hostEntity.getPort());
                rpcCallLog.setTargetAppName(hostEntity.getAppName());
            }
            rpcCallLog.setEnd(System.currentTimeMillis());
            rpcCallLog.setRt(rpcCallLog.getEnd() - rpcCallLog.getStart());
            rpcCallLog.setErrorMessage(throwable.getMessage());
            rpcCallLogHolder.log(rpcCallLog);
        }
    }

    private String createErrorMessage(NodeHostEntity hostEntity, RpcRequestBean rpcRequestBean) {
        StringBuilder builder = new StringBuilder("Error. ");
        if (null != rpcRequestBean) {
            builder.append(
                    "Invoke "
                            + rpcRequestBean.getFullClassName()
                            + "#"
                            + rpcRequestBean.getMethodName()
                            + "("
                            + StringUtils.join(rpcRequestBean.getArgumentTypes(), ",")
                            + ")@");
        }
        if (null != hostEntity) {
            builder.append(hostEntity.getHostInfo());
        }
        if (null != rpcRequestBean) {
            builder.append(
                    ". Arguments = " + JsonUtils.toJsonString(rpcRequestBean.getArgumentValues()));
        }
        return builder.toString();
    }

    private Object doInternalSendWhenCircuitBreakerClosed(
            NodeHostEntity hostEntity,
            RpcRequestBean rpcRequestBean,
            RpcCallLog rpcCallLog,
            RpcProtocol rpcProtocol) {
        rpcCallLog.setRequestId(rpcRequestBean.getId());
        rpcCallLog.setTargetIp(hostEntity.getIp());
        rpcCallLog.setTargetPort(hostEntity.getPort());
        rpcCallLog.setTargetAppName(hostEntity.getAppName());

        ReadClientFuture future = internalSend(rpcProtocol, hostEntity);
        RpcResponseBean rpcResponseBean = future.get();
        rpcCallLog.setEnd(System.currentTimeMillis());
        rpcCallLog.setRt(rpcCallLog.getEnd() - rpcCallLog.getStart());
        rpcCallLog.setErrorMessage(
                (null == rpcResponseBean || rpcResponseBean.getResult() instanceof RpcResponseError)
                        ? "Server ERROR"
                        : "");
        if (null == rpcResponseBean) {
            throw new ResultWasNullException("result is null.");
        }
        Object result = rpcResponseBean.getResult();
        if (result instanceof RpcResponseError) {
            RpcResponseError responseError = (RpcResponseError) result;
            throw new ServiceProviderOccurredException(
                    responseError.getCode() + ": " + responseError.getMessage());
        }
        rpcCallLogHolder.log(rpcCallLog);
        return result;
    }

    private String getCallLogKey(RpcRequestBean rpcRequestBean, NodeHostEntity hostEntity) {
        return rpcRequestBean.getFullClassName()
                + "."
                + rpcRequestBean.getMethodName()
                + (null == hostEntity ? "" : "@" + hostEntity.getHostInfo());
    }

    private ReadClientFuture internalSend(RpcProtocol rpcProtocol, NodeHostEntity hostEntity) {
        RpcClient rpcClient = rpcClients.getRpcClient(hostEntity);
        return rpcClient.send(rpcProtocol);
    }

    private NodeHostEntity chooseNodeHostEntity(String className, String methodFullName) {
        return nodeHostContainer.choose(className, methodFullName);
    }

    private RpcCallLog initRpcCallLog(
            RegisterBeanMethod registerBeanMethod,
            String className,
            Object[] args,
            String previousStackTraceElement) {
        RpcCallLog rpcCallLog = new RpcCallLog();
        rpcCallLog.setTargetClassName(className);
        rpcCallLog.setTargetMethodName(registerBeanMethod.toString());
        SerialNumberThreadLocal.TraceEntity traceEntity = SerialNumberThreadLocal.get();
        rpcCallLog.setSerialNumber(traceEntity.getSerialNumber());
        rpcCallLog.setCaller(traceEntity.getCaller());
        rpcCallLog.setRequestId(traceEntity.getCurrentRequestId());
        rpcCallLog.setThreadId(Thread.currentThread().getId());
        rpcCallLog.setStart(System.currentTimeMillis());
        rpcCallLog.setRequestArgs(args);
        rpcCallLog.setPreviousStackTrace(previousStackTraceElement);
        return rpcCallLog;
    }

    private RpcProtocol createRpcProtocol(RpcRequestBean requestBean) {
        // 封装 RpcProtocol
        RpcProtocol rpcProtocol = new RpcProtocol();
        rpcProtocol.setType(ConstantValues.BIZ_TYPE);
        rpcProtocol.setCmdId(IdGenHelper.getId());
        rpcProtocol.setData(serializerHolder.getSerializer().serialize(requestBean));
        return rpcProtocol;
    }

    private RpcRequestBean createRpcRequestBean(
            String className, RegisterBeanMethod registerBeanMethod, Object[] args) {
        // 封装 RpcRequestBean
        RpcRequestBean requestBean = new RpcRequestBean();
        requestBean.setFullClassName(className);
        requestBean.setMethodName(registerBeanMethod.getMethodName());
        requestBean.setArgumentTypes(registerBeanMethod.getParameterTypeNames());
        requestBean.setTs(System.currentTimeMillis());
        requestBean.setArgumentValues(args);
        SerialNumberThreadLocal.TraceEntity traceEntity = SerialNumberThreadLocal.get();
        requestBean.setId(traceEntity.getCurrentRequestId());
        requestBean.setSerialNumber(traceEntity.getSerialNumber());
        requestBean.setCaller(traceEntity.getCaller());
        return requestBean;
    }
}
